package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/15
 * @Desc: 学大数据，上多易教育
 *   各业务线，每5分钟的pv、uv和会话数
 **/
public class Job01_每5分钟各业务线流量 {

    public static void main(String[] args) {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中dwd层的行为日志topic: dwd-events
        tenv.executeSql("create table dwd_events_kafka(\n" +
                "     session_id         string\n" +
                "    ,event_id           string\n" +
                "    ,action_time        bigint\n" +
                "    ,user_id            bigint\n" +
                "    ,page_service       string\n" +
                "    ,rt as to_timestamp_ltz(action_time,3)         \n" +
                "    ,watermark for rt as rt - interval '0' second  \n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-events',\n" +
                "    'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "    'properties.group.id' = 'doit44_g1',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'value.format' = 'json',\n" +
                "    'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")                                                    ");

        // 建表，映射mysql中的看板指标结果表
        tenv.executeSql(
                        " create table sink_mysql(                       "+
                        "   page_service string,                         "+
                        " 	window_start timestamp(3),                   "+
                        " 	window_end   timestamp(3),                   "+
                        " 	pv           bigint,                         "+
                        " 	uv           bigint,                         "+
                        " 	ses_cnt      bigint,                         "+
                        " 	primary key(page_service) not enforced       "+
                        " ) with (                                       "+
                        "     'connector'='jdbc',                        "+
                        " 	'url' = 'jdbc:mysql://doitedu:3306/doit44',  "+
                        " 	'username' = 'root',                         "+
                        " 	'password' = 'root',                         "+
                        " 	'table-name' = 'w_5m_traffic_01'             "+
                        " )                                              "
        );

        // insert ...  统计sql
        tenv.executeSql(
                        " INSERT INTO sink_mysql                                      "+
                        " WITH tmp AS (                                               "+
                        "     SELECT                                                  "+
                        "         *                                                   "+
                        "     FROM dwd_events_kafka                                   "+
                        "     WHERE event_id = 'page_load'                            "+
                        " )                                                           "+
                        " SELECT                                                      "+
                        "    page_service,                                            "+
                        "    window_start,                                            "+
                        "    window_end,                                              "+
                        "    count(1) as pv,                                          "+
                        "    count(distinct user_id) as uv,                           "+
                        "    count(distinct session_id) as ses_cnt                    "+
                        " FROM TABLE(                                                 "+
                        "     TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '5' MINUTE)    "+
                        " )                                                           "+
                        " GROUP BY                                                    "+
                        "     window_start,                                           "+
                        " 	window_end,                                               "+
                        " 	page_service                                              "
        );
    }
}
